package com.atguigu.gmall.realtime.app.dws.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.TradeSkuOrderBean;
import com.atguigu.gmall.realtime.util.DimUtil;
import com.atguigu.gmall.realtime.util.ThreadPoolUtil;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

public class DimAsyncFuntion extends RichAsyncFunction<TradeSkuOrderBean,TradeSkuOrderBean> {

    //    2.1  一般情况查数据库的访问方式都是同步的
    //    2.2  构建线程池
    //    2.3   把各种请求提交到线程池，进行处理
    //    2.4   如果处理完成，在线程的任务中调用flink提供的
    //resultFuture.complete(Collections.singleton(结果));
    //resultFuture 是回调接口，用于通知Flink该条处理完成，并返回结果。
    @Override
    public void asyncInvoke(TradeSkuOrderBean tradeSkuOrderBean, ResultFuture<TradeSkuOrderBean> resultFuture) throws Exception {

        //获得连接池
        ThreadPoolExecutor threadPoolExecutor = ThreadPoolUtil.getPool();

        threadPoolExecutor.submit(new Runnable() {
            @Override
            public void run() {
                //
                System.out.println("线程："+ Thread.currentThread()+  "异步查询维度....");

                //维度查询
                String skuId = tradeSkuOrderBean.getSkuId();
                JSONObject dimSkuInfo = DimUtil.getDimInfoWithCache("dim_sku_info", skuId);

                tradeSkuOrderBean.setTrademarkId(dimSkuInfo.getString("tm_id"));
                tradeSkuOrderBean.setSpuId(dimSkuInfo.getString("spu_id"));
                tradeSkuOrderBean.setCategory3Id(dimSkuInfo.getString("category3_id"));

                System.out.println("线程："+ Thread.currentThread()+  "异步查询维度完成..");
                resultFuture.complete(Collections.singleton(tradeSkuOrderBean));
            }
        });

    }
}
